package org.yonggan.dmp.tag

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.yonggan.dmp.conf.ConfigManager
import org.yonggan.dmp.utils.{JedisPoolUtils, UserTagUtils}

/**
  * 处理用户标签
  * 过滤数据
  */
object UserTag {

  def main(args: Array[String]): Unit = {

    val config = new SparkConf()
      .setMaster("local[*]")
      .setAppName("处理用户标签")
    val sc = new SparkContext(config)
    val sqlContext = new SQLContext(sc)

    /**
      * 字典文件
      * 数据处理完了之后要记得收集到driver端，这样广播出去的数据才是全量的数据
      */
    val appdict = sc.textFile(ConfigManager.APP_DICT_FILE_PATH)
      .map(_.split("\t", -1))
      .filter(_.length >= 5)
      .map(arr => (arr(4), arr(1)))
      .collectAsMap()

    // 将数据广播出去
    val appbct = sc.broadcast(appdict)

    /**
      * 处理用户搜索停用词
      */
    val stopWord = sc.textFile(ConfigManager.USER_STOP_WORD)
    val stopWordArr = stopWord.collect()
    val stopWordBct: Broadcast[Array[String]] = sc.broadcast(stopWordArr)

    // 读取parqut 文件
    val pqDF = sqlContext.read.parquet(ConfigManager.PARQUET_OUT)

    // 提取用户数据
    val baseRDD = pqDF.rdd.mapPartitions(itor => {

      val jedis = JedisPoolUtils.getJedis(8)
      val resultItor: Iterator[(String, List[(String, Int)])] = itor.map(row => {
        // 用户身份表示
        val uid = UserTagUtils.userIdentity(row)
        // 广告类标签
        val adtag = AdTag.makeTag(row,jedis)
        // 设备标签
        val equip = EquipmentTag.makeTag(row)
        //关键字
        val keywordList = UserTagUtils.keyWordTag(row, stopWordBct)
        //        keywordList.foreach(println)
        (uid, adtag++equip++keywordList)
      })
      jedis.close()
      resultItor
    }).filter(!_._1.equals("无法标识用户"))

    //    baseRDD.foreach(println)
    //用户标签合并
    baseRDD.reduceByKey((lst1, lst2) => {
      //行为数据合并
      (lst1 ++ lst2)
        .groupBy(_._1)
        .mapValues(_.map(_._2).sum).toList
    }).foreach(println)

    sc.stop()
  }
}
